package com.cy.order.service;

import com.cy.order.entity.AccountMessage;
import com.cy.order.entity.Order;
import com.cy.order.entity.TxInfo;
import com.cy.order.feign.AccountClient;
import com.cy.order.feign.EasyIdClient;
import com.cy.order.feign.StorageClient;
import com.cy.order.mapper.OrderMapper;
import com.cy.order.mapper.TxMapper;
import com.cy.order.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.UUID;

@Service
@RocketMQTransactionListener
@Slf4j
public class OrderServiceImpl implements OrderService, RocketMQLocalTransactionListener {

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private AccountClient accountClient;

    @Autowired
    private StorageClient storageClient;

    @Autowired
    private EasyIdClient easyIdClient;

    //该对象封装了RocketMQ方法
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private TxMapper txMapper;

    //完成业务处理的方法
    public void doCreate(Order order) {

        //生成订单id
        String s = easyIdClient.nextId("order_business");
        Long orderId = Long.valueOf(s);
        order.setId(orderId);

        orderMapper.create(order);

        //rocketmq消息服务处获取消息，触发业务，不再通过远程调用执行业务
        //调用商品库存减少商品库存
        //storageClient.decrease(order.getProductId(),order.getCount());
        //调用账户金额减少账户金额
        //accountClient.decrease(order.getUserId(),order.getMoney());

    }

    // 业务方法中，不直接执行业务处理，
    // 而是发送事务消息，即半消息
    @Override
    public void create(Order order) {
        //用UUID随机产生一个事务xid
        String xid = UUID.randomUUID().toString().replaceAll("-", "");
        //用AccountMessage 封装数据
        AccountMessage amsg = new AccountMessage(order.getUserId(), order.getMoney(), xid);
        //转成json
        //"{userId:8,money:33.22,xid:y4y54}"
        String json = JsonUtil.to(amsg);
        //json封装到 spring 的 通用 message
        Message<String> msg = MessageBuilder.withPayload(json).build();

        //参数：1.发送地址：topic名：tag名；2.消息message；3.业务数据-订单对象
        rocketMQTemplate.sendMessageInTransaction(
                //如果有tag标签，格式为： orderTopic:tag1
                "orderTopic",
                msg,
                order);
    }

    //控制本地事务，保证业务的数据库事务和事务表事务的一致性
    @Transactional
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(
            Message message,
            //该Object类参数即为发送事务信息携带的业务数据对象 - order
            Object o) {
        // 存储到数据表的状态值用 0,1 表示
        int status;
        // 返回的事务状态是 State 对象
        RocketMQLocalTransactionState state;

        Order order = null;
        try {
            //执行原来的业务代码
            System.out.println("============执行本地事务==============");
            order = (Order) o;
            doCreate(order);

            //测试回查
            //Thread.sleep(70*1000);
            System.out.println("===========本地事务执行成功=============");

            status = 0;
            state = RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("本地事务存储订单失败",e);
            status = 1;
            state = RocketMQLocalTransactionState.ROLLBACK;
        }
        //把事务状态存到数据表的 tx_table
        //"{userId:8,money:33.22,xid:y4y54}"
        String json = new String((byte[]) message.getPayload());
        String xid = JsonUtil.getString(json, "xid");

        txMapper.insert(new TxInfo(xid,status,System.currentTimeMillis()));

        return state;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {

        System.out.println("==============正在回查=================");

        //通过message中的xid查询事务状态表中的status状态
        String json = new String((byte[]) message.getPayload());
        String xid = JsonUtil.getString(json, "xid");

        TxInfo txInfo = txMapper.selectById(xid);
        if (txInfo == null){
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        switch (txInfo.getStatus()){
            case 0: return RocketMQLocalTransactionState.COMMIT;
            case 1: return  RocketMQLocalTransactionState.ROLLBACK;
            default: return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
